package io.mft.as.ar.gateway.pool;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantLock;

import io.mft.as.ar.gateway.client.NetAddress;
import io.mft.as.ar.gateway.client.Server;
import io.mft.as.ar.gateway.pool.client.PoolAsClient;
import io.mft.as.ar.gateway.pool.client.PoolAsClientV2;
import io.mft.as.ar.gateway.util.SocketHelper;
import lombok.extern.slf4j.Slf4j;

/**
 * 描述:
 * 版权:	 Copyright (c) 2007
 * 公司:	 思迪科技
 * 作者:	 易庆锋
 * 版本:	 1.0
 * 创建日期: 2007-11-20
 * 创建时间: 17:16:58
 */
@Slf4j
public final class BusPool extends Thread
{
    
    
    //连接网络主机
    private String                            host;
    
    //连接端口
    private int                               port;
    
    //池数量
    private int                               poolSize;
    
    //调用Bus接口连接超时时间
    private int                               recvTimeout;
    
    //ThinkBus加密的key，如果BUS接口配置了key，这里必须与接口中配置的一致
    private String                            encryptKey  = "";
    
    //asClient队列
    private LinkedBlockingQueue<PoolAsClient> asQueue     = new LinkedBlockingQueue<PoolAsClient>();
    
    //    private List<PoolAsClient> asQueue     = new ArrayList<PoolAsClient>();
    
    //保存所有的as
    //    private ArrayList                         asList      = new ArrayList();
    
    //池连接检测线程
    //    private Thread             checkThread = null;
    
    //连接池当前是否可用
    private boolean                           isUseable   = false;
    
    //网关服务器名称
    private String                            gatewayName = "";
    
    private Server                            server      = null;
    
    //是否退出当前线程
    private boolean                           isStop      = false;
    
    private String                            version     = null;
    
    private final ReentrantLock               takelock    = new ReentrantLock();
    
    public BusPool(Server server, String gatewayName, String host, int port, String version)
    {
        this(server, gatewayName, host, port, 10, version);
    }
    
    /**
     * 创建线程池
     *
     * @param host
     * @param port
     * @param poolSize
     */
    public BusPool(Server server, String gatewayName, String host, int port, int poolSize, String version)
    {
        this.server = server;
        this.gatewayName = gatewayName;
        this.host = host;
        this.port = port;
        this.poolSize = poolSize;
        this.version = version;
        initPool();
    }
    
    public BusPool(Server server, String gatewayName, String host, int port, int poolSize, int recvTimeout,
            String encryptKey, String version)
    {
        this.server = server;
        this.gatewayName = gatewayName;
        this.host = host;
        this.port = port;
        this.poolSize = poolSize;
        this.recvTimeout = recvTimeout;
        this.encryptKey = encryptKey;
        this.version = version;
        initPool();
        
    }
    
    public void startCheckPoolExpire()
    {
        CheckPoolExpire expire = new CheckPoolExpire();
        expire.start();
    }
    
    /**
    * 初始化连接池
    */
    public void initPool()
    {
        try
        {
            if ( SocketHelper.connectServer(host, port) )
            {
                //                for (int i = 0; i < poolSize; i++)
                //                {
                //                    createPool();
                //                }
                isUseable = true;
                log.info("初始化BUS连接池[" + host + ":" + port + "]成功");
            }
            else
            {
                isUseable = false;
            }
        }
        catch (Exception ex)
        {
            log.error("初始化BUS连接池[" + host + ":" + port + "]失败");
        }
    }
    
    /**
     * 
     * 描述：创建连接池
     * 作者：李炜
     * 时间：Dec 9, 2013 11:12:49 PM
     */
    public void createPool()
    {
        if ( asQueue.size() < poolSize )
        {
            try
            {
                PoolAsClient socket = null;
                if("2".equals(version))
                {
                    socket = new PoolAsClientV2(this.host, this.port, this.recvTimeout, this.encryptKey, this);
                }
                //                asList.add(socket);
                asQueue.put(socket);
            }
            catch (InterruptedException e)
            {
                log.error(e.getMessage(), e);
            }
            catch (Exception e)
            {
                log.error(e.getMessage(), e);
            }
        }
    }
    
    /**
     * 
     * 描述：关闭所有连接池
     * 作者：李炜
     * 时间：Dec 9, 2013 11:14:20 PM
     */
    public void closeAllPool()
    {
        isUseable = false;
        Iterator<PoolAsClient> it = asQueue.iterator();
        while (it.hasNext())
        {
            PoolAsClient asClient = it.next();
            try
            {
                asClient.setIsUsable(false);
                asClient.close();
            }
            catch (Exception ex)
            {
                log.error(ex.getMessage(), ex);
            }
        }
        asQueue.clear();
    }
    
    public boolean isUseable()
    {
        return isUseable;
    }
    
    public void setUseable(boolean isUseable)
    {
        this.isUseable = isUseable;
    }
    
    public PoolAsClient getClient()
    {
        PoolAsClient client = null;
        if ( !isUseable ) //若不可用，则返回空的asClient对象
        {
            return client;
        }
        
        try
        {
            //20170220 modify by liwei 如果池子中对象为空，则等待连接释放
            /*
            client = (PoolAsClient) asQueue.poll(10, TimeUnit.SECONDS); //若取不到，则等待10秒钟再取
            if ( client == null )//取不到，创建一个短连接
            {
                PoolAsClient poolAsClient = new PoolAsClient(host, port);
                poolAsClient.setIsUsable(true);
                return poolAsClient;
            }
            */
            //先判断连接池是否有连接
            
            if ( asQueue.isEmpty() )
            {
                createPool();
            }
            
            client = asQueue.take();
        }
        catch (Exception ex)
        {
            log.error("", ex);
        }
        return client;
        
    }
    
    public String getHost()
    {
        return this.host;
    }
    
    public int getPort()
    {
        return this.port;
    }
    
    public int getPoolSize()
    {
        return poolSize;
    }
    
    public int getAsListSize()
    {
        return asQueue.size();
    }
    
    public ArrayList getAsList()
    {
        ArrayList<PoolAsClient> asList = new ArrayList<PoolAsClient>();
        Iterator<PoolAsClient> it = asQueue.iterator();
        while (it.hasNext())
        {
            PoolAsClient asClient = it.next();
            asList.add(asClient);
        }
        return asList;
    }
    
    public int getQueueSize()
    {
        return asQueue.size();
    }
    
    public LinkedBlockingQueue<PoolAsClient> getAsQueue()
    {
        return asQueue;
    }
    
    public boolean offer(PoolAsClient client)
    {
        return asQueue.offer(client);
    }
    
    /**
     * 
     * @描述: 判断当前IP地址和端口是否在配置文件BusConfig.xml中存在，如果不存在，返回false
     * @作者: liwei
     * @创建日期: 2016年2月10日 下午12:48:07
     * @return
     */
    private boolean isExists()
    {
        boolean isExists = false;
        if ( server != null )
        {
            NetAddress[] netAddressArr = server.getNetAddressArr();
            if ( netAddressArr != null && netAddressArr.length > 0 )
            {
                for (int i = 0; i < netAddressArr.length; i++)
                {
                    NetAddress netAddress = netAddressArr[i];
                    if ( netAddress != null && this.getHost().equals(netAddress.getIP())
                            && this.port == netAddress.getPort() )
                    {
                        isExists = true;
                        break;
                    }
                }
            }
        }
        return isExists;
    }
    
    @Override
    public void run()
    {
        
        int count = 0;
        while (true)
        {
            try
            {
                sleep(1000); //每1秒钟检查一次
                boolean isExists = isExists();
                if ( !isExists )
                {
                    
                    //关闭所有连接池
                    closeAllPool();
                    this.server.removeAsPool(this);
                    log.error("网关服务器" + gatewayName + "[" + host + ":" + port + "]已退出");
                    break;
                }
                
                if ( SocketHelper.connectServer(host, port) ) //若服务器可以连接，则设置BusPool为可用状态
                {
                    setUseable(true);
                }
                else
                {
                    //循环3次，如果3次sokcet连接都失败，关半连接池中的所有连接
                    setUseable(false);
                    count += 1;
                    if ( count >= 3 )
                    {
                        count = 0;
                    }
                    else
                    {
                        continue;
                    }
                    
                    //判断所有连接池
                    closeAllPool();
                    
                    //每隔1秒检查一次服务器是否已经可以连接
                    while ( !SocketHelper.connectServer(host, port))
                    {
                        log.error("网关服务器" + gatewayName + "[" + host + ":" + port + "]是否存活=false");
                        isExists = isExists();
                        if ( !isExists )
                        {
                            isStop = true;
                            break;
                        }
                        sleep(1000);
                    }
                    if ( isStop )
                    {
                        log.error("网关服务器" + gatewayName + "[" + host + ":" + port + "]已退出");
                        this.server.removeAsPool(this);
                        break;
                    }
                    
                    //服务器已经可以连接，则重新初始化池
                    initPool();
                    
                }
                
            }
            catch (Exception ex)
            {
                log.error("", ex);
            }
            
        }
        
    }
    
    class CheckPoolExpire extends Thread
    {
        public CheckPoolExpire()
        {
            
        }
        
        public void run()
        {
            long checktime = server.getCheckPoolExpired();//间隔检查时间
            long scantime = server.getScanAliveInterval();//闲置最大时间，超过此时间将从队列中释放此连接
            if ( checktime <= 0 )
            {
                checktime = 60000;
            }
            while (true)
            {
                try
                {
                    Thread.sleep(checktime);
                }
                catch (InterruptedException e)
                {
                    log.error("", e);
                }
                
                //扫描存活时长没有配置或者配置为0则不检查
                if ( server != null )
                {
                    if ( scantime == 0 )
                    {
                        continue;
                    }
                    //连接池没有连接则跳过
                    try
                    {
                        if ( null != asQueue && !asQueue.isEmpty() )
                        {
                            long current = System.currentTimeMillis();
                            int length = asQueue.size();
                            for (int i = 0; i < length; i++)
                            {
                                PoolAsClient poolAsClient = asQueue.poll();
                                if ( null == poolAsClient )
                                {
                                    continue;
                                }
                                long duration = current - poolAsClient.getTime();
                                if ( duration > scantime )
                                {
                                    if ( log.isDebugEnabled() )
                                    {
                                        log.debug("验证ScanAliveInterval---DURATIONS:" + duration);
                                    }
                                    asQueue.remove(poolAsClient);
                                    poolAsClient.setIsUsable(false);
                                    poolAsClient.close();
                                }
                                else
                                {
                                    asQueue.offer(poolAsClient);
                                }
                            }
                        }
                    }
                    catch (Exception e)
                    {
                        log.error("", e);
                    }
                }
            }
        }
    }
    
}
